package org.example.flink.windows;


import org.apache.commons.lang3.RandomStringUtils;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.example.flink.util.KafkaUtil;

public class Hello12WatermarkGenerator {
    public static void main(String[] args) throws Exception {
        //启动一个线程专门发送消息给Kafka，这样我们才有数据消费
        new Thread(() -> {
            String uname = RandomStringUtils.randomAlphabetic(8);
            for (int i = 1000; i < 2000; i++) {
                KafkaUtil.sendMsg("yjxxt", uname + i % 2 + ":" + i + ":" + System.currentTimeMillis());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        //获取环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        //设置Kafka连接
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.101.100:9092,192.168.101.101:9092,192.168.101.102:9092")
                .setTopics("yjxxt")
                .setGroupId("flink_KafkaConnector")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        //读取数据源
        DataStreamSource<String> kafkaSource =
                environment.fromSource(source, WatermarkStrategy.noWatermarks(), "KafkaSource");
        kafkaSource.assignTimestampsAndWatermarks(new YjxxtWatermarkStrategy())
                .keyBy(t -> t.split(":")[0])
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce((t1, t2) -> t1 + "[" + t2.split(":")[1] + "," + t2.split(":")[2] + "]")
                .map(t -> "[" + System.currentTimeMillis() + "][" + t + "]")
                .print("YjxxtWatermarkStrategy");
        //执行环境
        environment.execute();
    }
}
class YjxxtWatermarkStrategy implements WatermarkStrategy<String> {
    @Override
    public WatermarkGenerator<String> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new YjxxtPeriodicGenerator();
    }
    @Override
    public TimestampAssigner<String> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return new SerializableTimestampAssigner<String>() {
            @Override
            public long extractTimestamp(String element, long recordTimestamp) {
                return Long.valueOf(element.split(":")[2]);
            }
        };
    }
    /**
     * 周期型生成水位线
     */
    private static class YjxxtPeriodicGenerator implements WatermarkGenerator<String> {
        // 延迟时间
        private Long lateTime = 3000L;
        // 观察到的最大时间戳
        private Long maxTimestamp = Long.MIN_VALUE;
        @Override
        public void onEvent(String element, long eventTimestamp, WatermarkOutput output) {
            maxTimestamp = Math.max(Long.valueOf(element.split(":")[2]), maxTimestamp); //
        }
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            output.emitWatermark(new Watermark(maxTimestamp - lateTime - 1L));
        }
    }
    /**
     * 定点型生成水位线
     */
    private static class YjxxtPunctuatedGenerator implements WatermarkGenerator<String> {
        @Override
        public void onEvent(String element, long eventTimestamp, WatermarkOutput output) {
            if (Integer.parseInt(element.split(":")[1]) % 100 == 0) {
                output.emitWatermark(new Watermark(Long.valueOf(element.split(":")[2]) - 1L));
            }
        }
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
        }
    }
}